package main.java.com.lee.pattern.masterworker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.IntStream;

/**
 * @Description TODO
 * @Author winston
 * @DateTime 2022/2/18
 */
public class Master<T extends Task, R> {
    private final Map<String, Worker<T, R>> workers = new HashMap<>();
    private final LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    protected Map<String, R> resultMap = new ConcurrentHashMap<>();
    private final AtomicLong sum = new AtomicLong(0);

    Thread thread = null;

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 10, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));

    public Master(int workerCount) {
        IntStream.range(0, workerCount).forEach(i -> {
            Worker<T, R> worker = new Worker<>(executor);
            workers.put("worker: " + i, worker);
        });
        thread = new Thread(this::execute);
        thread.start();
    }

    public void execute() {
        for (; ; ) {
            if (Thread.currentThread().isInterrupted()) {
                System.out.println("主线程被中断");
                break;
            }
            workers.forEach((key, worker) -> {
                T t;
                try {
                    t = this.taskQueue.take();
                    worker.submit(t, this::resultCallBack);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }



    public void submit(T t) {
        taskQueue.add(t);
    }

    private void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker:" + task.getWorkerId() + "-" + "Task: " + task.getTaskId();
        R result = task.getResult();
        resultMap.put(taskName, result);
        sum.getAndAdd((Integer) result);
    }

    public void printResult() {
        System.out.println("sum is " + sum.get());
        resultMap.forEach((k, v) -> {
            System.out.println(k + " : " + v);
        });
    }

    public void stop() {
        executor.shutdown();
        thread.interrupt();
        workers.values().forEach(i -> {
            if (i.getThread() != null) {
                i.getThread().interrupt();
                System.out.println("中断线程池线程"+i.getThread().getName());
            }
        });
        while(!executor.isTerminated()){
            try {
                System.out.println(executor.awaitTermination(3, TimeUnit.SECONDS));
                System.out.println(executor.getActiveCount());
                Thread.sleep(1000);
                System.out.println("循环关闭线程池");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executor.shutdownNow();
        if(executor.isShutdown()){
            System.out.println("executor is shutdown");
        }else {
            System.out.println("executor is not shutdown");
        }
    }
}
